/**
 * 聚合函数
 *
 */
package com.atguigu.day8

import com.atguigu.source.{SensorReading, SensorSource}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.scala._
import org.apache.flink.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.table.api._
import org.apache.flink.table.functions.AggregateFunction
import org.apache.flink.types.Row


object AggregateFunctionExample {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val settings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()


    val tEnv = StreamTableEnvironment.create(env,settings)

    val stream = env.addSource(new SensorSource)
    var avgTemp = new AvgTemp()
    val table = tEnv.fromDataStream(stream,'id,'timeStamP,'temperature)



    //table api写法
    table
      .groupBy('id)
      .aggregate(avgTemp('temperature) as 'avgTemp)
      .select('id,'avgTemp)
      .toRetractStream[Row]
//      .print()

    //sql写法
    tEnv.createTemporaryView("sensor",table)
    tEnv.registerFunction("avgTemp",avgTemp)
    tEnv
        .sqlQuery(
          """
            |SELECT
            |id,avgTemp(temperature)
            |FROM
            |sensor
            |GROUP BY id
            |""".stripMargin)
        .toRetractStream[Row]
        .print()


    env.execute()
  }

  //累加器类型
  class AvgTempAcc{

    var sum:Double = 0.0
    var count:Int = 0
  }
//第一个泛型是温度值的类型
//第二个泛型是累加器的类型
//累加器规则必须是  accumulate  名字
  class AvgTemp extends AggregateFunction[Double,AvgTempAcc]{

  //创建累加器
  override def createAccumulator(): AvgTempAcc = new AvgTempAcc

  //累加器规则
  def accumulate(acc:AvgTempAcc,temp:Double): Unit ={
    acc.sum += temp
    acc.count += 1
  }



  override def getValue(acc: AvgTempAcc): Double = {
    acc.sum / acc.count
  }

}

}
